package com.bw.gmall.realtime.dws_new6;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.utils.MyClickHouseUtil;
import com.bw.gmall.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.*;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import org.apache.flink.api.java.tuple.Tuple9;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

public class DwsTradeWarningOrderByShop {
    public static void main(String[] args) throws Exception {
        // 1. 获取流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //TODO 2.读取DWD层成功支付主题数据创建流
        String topic = "dwd_trade_order_refund_a";
        String groupId = "dwd_trade_order_refund_a_250708";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));


        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {
                return JSONObject.parseObject(value);
            }
        });


        SingleOutputStreamOperator<JSONObject> withWatermarkStream = jsonStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return java.sql.Timestamp.valueOf(element.getString("create_time")).getTime();
                            }
                        })
        );

        // 4. 提取关键信息：(店铺ID, 退款类型, 退款金额, 处理中标记, 成功标记, 失败标记, 申请中, 可挽回)
          // 修复：替换 Lambda 为匿名内部类，显式声明泛型
        SingleOutputStreamOperator<Tuple8<String, String, Double, Integer, Integer, Integer,Integer,Integer>> refundTupleStream = withWatermarkStream.map(
                new MapFunction<JSONObject, Tuple8<String, String, Double, Integer, Integer, Integer,Integer,Integer>>() {
                    @Override
                    public Tuple8<String, String, Double, Integer, Integer, Integer,Integer,Integer> map(JSONObject value) throws Exception {
                        String shopId = value.getString("shop_id");
                        String refundType = value.getString("order_status");
                        Double amount = Double.parseDouble(value.getString("refund_amount"));

                        int isProcessing = 0;
                        int isSuccess = 0;
                        int isFail = 0;
                        int isapplication = 0;
                        int isrecoverable=0;


                        if ("1003".equals(refundType)) {
                            isSuccess = 1;
                        } else if ("1005".equals(refundType)) {
                            isFail = 1;
                        } else  if("1002".equals(refundType)) {
                            isProcessing = 1;
                        } else if ("1001".equals(refundType)) {
                            isapplication =1;
                        }else if  ("1004".equals(refundType)){
                            isrecoverable=1;
                        }

                        return new Tuple8<>(shopId, refundType, amount, isProcessing, isSuccess, isFail,isapplication,isrecoverable);
                    }
                }
        );
        refundTupleStream.print();

// 定义结果数据流：Tuple11包含11个字段，分别对应：
// 按店铺ID分组，统计所有状态
        SingleOutputStreamOperator<Tuple11<String, String, Double, Double, Double, Integer, Integer, Integer, Integer, String, String>> resultStream = refundTupleStream
                .keyBy(new KeySelector<Tuple8<String, String, Double, Integer, Integer, Integer, Integer, Integer>, String>() {
                    @Override
                    public String getKey(Tuple8<String, String, Double, Integer, Integer, Integer, Integer, Integer> value) throws Exception {
                        return value.f0; // 按店铺ID分组
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.days(1)))
                .aggregate(
                        new AggregateFunction<
                                Tuple8<String, String, Double, Integer, Integer, Integer, Integer, Integer>,
                                // 注意：Tuple9的索引是f0~f8，共9个字段
                                Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer>,
                                Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer>>() {

                            @Override
                            public Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> createAccumulator() {
                                return new Tuple9<>("", "", 0.0, 0.0, 0.0, 0, 0, 0, 0);
                            }

                            @Override
                            public Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> add(
                                    Tuple8<String, String, Double, Integer, Integer, Integer, Integer, Integer> value,
                                    Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> accumulator) {

                                // 累加总金额
                                double totalAmount = accumulator.f2 + value.f2;
                                // 累加成功金额（仅当状态为成功时）
                                double successAmount = accumulator.f3 + (value.f4 == 1 ? value.f2 : 0);
                                // 累加失败金额（仅当状态为失败时）
                                double failAmount = accumulator.f4 + (value.f5 == 1 ? value.f2 : 0);
                                // 累加各状态数量
                                int processingCount = accumulator.f5 + value.f3;
                                int successCount = accumulator.f6 + value.f4;
                                int failCount = accumulator.f7 + value.f5;
                                int recoverableCount = accumulator.f8 + value.f6; // 修正：使用f8作为可挽回数量

                                // 保留当前状态（后续会根据出现次数确定主要状态）
                                String status = value.f1;

                                return new Tuple9<>(
                                        value.f0,          // 店铺ID
                                        status,            // 状态值
                                        totalAmount,       // 总金额
                                        successAmount,     // 成功金额
                                        failAmount,        // 失败金额
                                        processingCount,   // 处理中数量
                                        successCount,      // 成功数量
                                        failCount,         // 失败数量
                                        recoverableCount   // 可挽回数量（修正：使用f8）
                                );
                            }

                            @Override
                            public Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> getResult(
                                    Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> merge(
                                    Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> a,
                                    Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> b) {

                                // 合并相同店铺的统计结果
                                return new Tuple9<>(
                                        a.f0,              // 店铺ID
                                        a.f1,              // 状态值（后续会处理）
                                        a.f2 + b.f2,       // 合并总金额
                                        a.f3 + b.f3,       // 合并成功金额
                                        a.f4 + b.f4,       // 合并失败金额
                                        a.f5 + b.f5,       // 合并处理中数量
                                        a.f6 + b.f6,       // 合并成功数量
                                        a.f7 + b.f7,       // 合并失败数量
                                        a.f8 + b.f8        // 合并可挽回数量（修正：使用f8）
                                );
                            }
                        },
                        new ProcessWindowFunction<
                                Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer>,
                                Tuple11<String, String, Double, Double, Double, Integer, Integer, Integer, Integer, String, String>,
                                String,
                                TimeWindow>() {

                            @Override
                            public void process(
                                    String shopId,
                                    Context context,
                                    Iterable<Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer>> elements,
                                    Collector<Tuple11<String, String, Double, Double, Double, Integer, Integer, Integer, Integer, String, String>> out) {

                                Tuple9<String, String, Double, Double, Double, Integer, Integer, Integer, Integer> aggResult = elements.iterator().next();
                                long windowStart = context.window().getStart();
                                long windowEnd = context.window().getEnd();

                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                String windowStartStr = sdf.format(new Date(windowStart));
                                String windowEndStr = sdf.format(new Date(windowEnd));

                                // 输出结果：保留店铺ID和状态值，以及各统计指标
                                out.collect(new Tuple11<>(
                                        aggResult.f0,      // 店铺ID
                                        aggResult.f1,      // 状态值
                                        aggResult.f2,      // 总金额
                                        aggResult.f3,      // 成功金额
                                        aggResult.f4,      // 失败金额
                                        aggResult.f5,      // 处理中数量
                                        aggResult.f6,      // 成功数量
                                        aggResult.f7,      // 失败数量
                                        aggResult.f8,      // 可挽回数量（修正：使用f8）
                                        windowStartStr,    // 窗口开始时间
                                        windowEndStr       // 窗口结束时间
                                ));
                            }
                        }
                );


// 6. 输出结果
        resultStream.print("各店铺退款统计：(店铺ID, 状态类型, 退款总金额, 退款成功金额, 退款失败金额, 处理中退款个数, 退款失败个数,申请中的个数,可挽回的个数, 窗口开始, 窗口结束)");
//------------------------------------------------
// 修改 Sink SQL，明确字段和占位符，与 ClickHouse 表结构完全匹配
        resultStream.addSink(MyClickHouseUtil.getSinkFunction_a(
                "INSERT INTO dws_refund_shop_stats " +
                        "VALUES (?,?,?,?,?,?,?,?,?,?,?)"
        ));


        // 9. 执行任务
        env.execute("DWS - Calculate Refund Total Amount By Shop");





    }
}
